﻿using NetSync.Core;
using Newtonsoft.Json.Linq;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

namespace NetSync.Rabbitmq
{
    public class NetRabbitMQConnection : INetConnection, IDisposable
    {
        private ConnectionFactory _factory1;
        private ConnectionFactory _factory2;
        private readonly IDataFilterService _filterService;
        private List<ConsumerInfo> _consumers;
        private string _name;
        private RabbitMQConfig _config;
        private string _exchangeName = "amqp.netsync.exchange";
        private bool _isWorkQueue = false;
        private SemaphoreSlim _semaphore;
        private IConnection[] _cConnections = null;
        private IConnection[] _pConnections = null;
        private int _curPIndex = 0;
        /// <summary>
        /// ctor
        /// </summary>
        public NetRabbitMQConnection(INetSerialize netSerialize, IDataFilterService filterService = null)
        {
            NetSerialize = netSerialize ?? throw new ArgumentNullException(nameof(netSerialize));
            _filterService = filterService;
        }

        public string Name => _name;

        public INetSerialize NetSerialize { get; private set; }

        public event EventHandler<ConnectionEventArgs> ConnectionEvent;

        public  event EventHandler<DispatchEventArgs> DispatchMessageEvent;

        public event EventHandler<DispatchErrorEventArgs> DispatchMessageErrorEvent;

        public async Task InitAsync(NetConfig config, CancellationToken token = default)
        {
            if (!(config is RabbitMQConfig c))
                throw new NetConnectionException($"{nameof(config)} is not{nameof(RabbitMQConfig)}) type");
            Dispose();
            InitConfig(c);
            _factory1 = GetFactory(true);
            _factory2 = GetFactory(false);
            _cConnections = new IConnection[_config.ConnectionCout];
            if (!_config.NewConnectionPerSend)
                _pConnections = new IConnection[_config.ConnectionCout];
            for(var i = 0; i < _config.ConnectionCout; i++)
            {
                var cconn = await CreateConnectionAsync(_config.Address, NetConncetionTypeEnum.Consumer.ToString(), token);
                _cConnections[i] = cconn;
                if (!_config.NewConnectionPerSend)
                {
                    var pconn = await CreateConnectionAsync(_config.Address, NetConncetionTypeEnum.Producer.ToString(), token);
                    _pConnections[i] = pconn;
                }
            }
        }

        public async Task PublishAsync(string topic, byte[] data, CancellationToken token = default)
        {
            if (topic == null) throw new ArgumentNullException(nameof(topic));
            if (data == null || data.Length == 0) throw new ArgumentNullException(nameof(data));
            await PublishCoreAsync(topic, data, token);
        }

        public async Task PublishAsync<TData>(string topic, TData data, CancellationToken token = default)
        {
            var byteData = await NetSerialize.SerializeAsync(data); 
            await PublishAsync(topic, byteData, token);
        }

        public async Task SubscribeAsync(List<SubscribeInfo> subscribeInfos, CancellationToken token = default)
        {
            if (subscribeInfos == null || subscribeInfos.Count == 0) throw new ArgumentNullException(nameof(subscribeInfos));
            await LockConsumerProcessAsync(subscribeInfos, token, async (s, t) =>
            {
                var groupConsumerInfos = s.Where(m => !string.IsNullOrWhiteSpace(m.Group)).GroupBy(m => m.Group);
                for(var j = 0; j < _cConnections.Length; j++)
                {
                    foreach (var info in groupConsumerInfos)
                    {
                        for (var i = 0; i < _config.ConsumerCountPerGroup; i++)
                        {
                            await SubscribeAsync(_cConnections[j], info, token);
                        }
                    }
                }
            });
        }

        public async Task UnsubscribeAsync(List<SubscribeInfo> subscribeInfos, CancellationToken token = default)
        {
            if (subscribeInfos == null || subscribeInfos.Count == 0) return;
            if (_consumers.Count == 0) return;
            //if (_cConnection?.IsOpen != true) throw new NetConnectionException("rabbitmq connection is not connected");
            await LockConsumerProcessAsync(subscribeInfos, token, async (s, t) =>
            {
                foreach (var item in s.GroupBy(m => m.Group))
                {
                    await UnsubscribeAsync(item, token);
                }
            });
        }

        public Delegate[] ListEventFunc(string eventName, bool reset = false)
        {
            Delegate[] delegates = null;
            switch (eventName)
            {
                case nameof(ConnectionEvent):
                    delegates = ConnectionEvent?.GetInvocationList() ?? new Delegate[0];
                    if (reset)
                        ConnectionEvent = null;
                    break;
                case nameof(DispatchMessageEvent):
                    delegates = DispatchMessageEvent?.GetInvocationList() ?? new Delegate[0];
                    if (reset)
                        DispatchMessageEvent = null;
                    break;
                case nameof(DispatchMessageErrorEvent):
                    delegates = DispatchMessageErrorEvent?.GetInvocationList() ?? new Delegate[0];
                    if (reset)
                        DispatchMessageErrorEvent = null;
                    break;
                default:
                    delegates = new Delegate[0];
                    break;
            }

            return delegates;
        }
        #region Dispose
        public void Dispose()
        {
            _semaphore?.Dispose();
            if (_pConnections != null)
            {
                foreach (var connection in _pConnections)
                {
                    DisposeConnection(connection);
                }
                _pConnections = null;
            }
            if (_cConnections != null)
            {
                foreach(var connection in _cConnections)
                {
                    DisposeConnection(connection);
                }
                _cConnections = null;
            }
            _factory1 = null;
            _factory2 = null;
            _config = null;
            _consumers?.Clear();
        }
        #endregion

        #region private methods
        private void InitConfig(RabbitMQConfig config)
        {
            if (config == null) throw new ArgumentNullException(nameof(config));
            if (string.IsNullOrWhiteSpace(config.Address)) throw new ArgumentNullException(nameof(config.Address));
            if (string.IsNullOrWhiteSpace(config.User)) throw new ArgumentNullException(nameof(config.User));
            if (string.IsNullOrWhiteSpace(config.Password)) throw new ArgumentNullException(nameof(config.Password));
            if (config.PrefetchCount < 1) config.PrefetchCount = 1;
            _config = config;
            _config.ConnectionCout = _config.ConnectionCout <= 0 ? 1 : _config.ConnectionCout;
            _name = config.Name;
            _consumers = new List<ConsumerInfo>();
            _semaphore = new SemaphoreSlim(1, 1);
            if (string.IsNullOrWhiteSpace(config.ExchangeName))
                _isWorkQueue = true;
            else
            {
                _isWorkQueue = false;
                _exchangeName = config.ExchangeName;
            }
        }

        private async Task<IConnection> CreateConnectionAsync(string address, string providerName, CancellationToken token)
        {
            var hosts = address.Split(new char[1] { ';' }, StringSplitOptions.RemoveEmptyEntries);
            var endpoints = new AmqpTcpEndpoint[hosts.Length];
            for (var i = 0; i < endpoints.Length; i++)
            {
                var hostname = hosts[i].AsSpan().Slice(0, hosts[i].IndexOf(':')).ToString();
                var port = int.Parse(hosts[i].AsSpan().Slice(hosts[i].IndexOf(':') + 1).ToString());
                var amqpEndpoint = new AmqpTcpEndpoint(hostname, port, _factory1.Ssl, _factory1.MaxInboundMessageBodySize);
                endpoints[i] = amqpEndpoint;
            }

            if (_config.NewConnectionPerSend && providerName == NetConncetionTypeEnum.Producer.ToString())
                return await _factory1.CreateConnectionAsync(endpoints, providerName, token);

            var connection = await _factory2.CreateConnectionAsync(endpoints, providerName, token);
            connection.ConnectionShutdownAsync += ShutdownEventAsync;
            connection.ConnectionBlockedAsync += BlockedEventAsync;
            connection.ConnectionUnblockedAsync += UnBlockedEventAsync;
            ConnectionEvent?.Invoke(connection, new ConnectionEventArgs { ConncetionType = ToNetConncetionType(connection), Connection = NetConnectionEnum.Connected });
            return connection;
        }

        private ConnectionFactory GetFactory(bool isP)
        {
            return isP ?
            new ConnectionFactory
            {
                AutomaticRecoveryEnabled = !_config.NewConnectionPerSend,
                MaxInboundMessageBodySize = _config.MaxMessageSize,
                UserName = _config.User,
                Password = _config.Password,
                VirtualHost = _config.VirtualPath,
                Ssl = _config.Ssl,
                RequestedConnectionTimeout = TimeSpan.FromSeconds(10),
            }
            :
            new ConnectionFactory
            {
                MaxInboundMessageBodySize = _config.MaxMessageSize,
                UserName = _config.User,
                Password = _config.Password,
                VirtualHost = _config.VirtualPath,
                Ssl = _config.Ssl,
                RequestedConnectionTimeout = TimeSpan.FromSeconds(10),
            };
        }

        private async Task<IChannel> CreateChannelAsync(IConnection connection, CancellationToken token)
        {
            if (connection?.IsOpen != true)
                throw new NetConnectionException("rabbitmq connection is not connected");
            return await connection.CreateChannelAsync(cancellationToken: token);
        }

        private AsyncEventingBasicConsumer CreateConsumerAsyc(IChannel channel, CancellationToken token)
        {
            var consumer = new AsyncEventingBasicConsumer(channel);
            consumer.ReceivedAsync += ConsumerReceiveEventAsync;
            return consumer;
        }

        private async Task ConsumerReceiveEventAsync(object consumer, BasicDeliverEventArgs ea)
        {
            var message = new DispatcherMessage(ea.RoutingKey, ea.Body, this);
            var errorFlag = false;
            try
            {
                //await DispatchMessageAsync(message);
                DispatchMessageEvent?.Invoke(this, new DispatchEventArgs(ea.RoutingKey, ea.Body, this, message.IsHandled));

            }
            catch (Exception e)
            {
                errorFlag = true;
                try
                {
                    DispatchMessageErrorEvent?.Invoke(this, new DispatchErrorEventArgs(ea.RoutingKey, ea.Body, this, e, message.IsHandled));
                }
                catch { }
            }
            finally
            {
                if (!errorFlag || _config.ReceiveHandleErrorAck)
                    await ((AsyncEventingBasicConsumer)consumer).Channel.BasicAckAsync(ea.DeliveryTag, false);
            }
        }

        private async Task LockConsumerProcessAsync(List<SubscribeInfo> subscribeInfos, CancellationToken token, Func<List<SubscribeInfo>, CancellationToken, Task> func)
        {
            try
            {
                await _semaphore.WaitAsync();
                await func.Invoke(subscribeInfos, token);
            }
            catch (Exception ex)
            {
                throw new NetConnectionException(ex.Message, ex);
            }
            finally
            {
                if (_semaphore?.CurrentCount == 0)
                    _semaphore?.Release();
            }
        }

        private void DisposeConnection(IConnection connection)
        {
            if (connection == null) return;
            try
            {
                connection.ConnectionShutdownAsync -= ShutdownEventAsync;
                connection.ConnectionBlockedAsync -= BlockedEventAsync;
                connection.ConnectionUnblockedAsync -= UnBlockedEventAsync;
                connection.CloseAsync().ConfigureAwait(false).GetAwaiter().GetResult();
                connection.Dispose();
                connection = null;
            }
            catch { }
        }

        private Task ShutdownEventAsync(object sender, ShutdownEventArgs e)
        {
            var eventArgs = new ConnectionEventArgs 
            { 
                ConncetionType = ToNetConncetionType((IConnection)sender),
                Connection = NetConnectionEnum.DisConnected,
                Message = $"{e.ClassId}|{e.ReplyCode}|{e.ReplyText}" 
            };
            ConnectionEvent?.Invoke(this, eventArgs);
            return Task.CompletedTask;
        }

        private Task BlockedEventAsync(object sender, ConnectionBlockedEventArgs e)
        {
            var eventArgs = new ConnectionEventArgs { ConncetionType = ToNetConncetionType((IConnection)sender), Connection = NetConnectionEnum.Blocked, Message = e.Reason };
            ConnectionEvent?.Invoke(this, eventArgs);
            return Task.CompletedTask;
        }

        private Task UnBlockedEventAsync(object sender, AsyncEventArgs e)
        {
            var eventArgs = new ConnectionEventArgs { ConncetionType = ToNetConncetionType((IConnection)sender), Connection = NetConnectionEnum.UnBlocked };
            ConnectionEvent?.Invoke(this, eventArgs);
            return Task.CompletedTask;
        }

        private NetConncetionTypeEnum ToNetConncetionType(IConnection conncetion)
        {
            if (conncetion == null)
                return NetConncetionTypeEnum.Unknown;
            if (Enum.TryParse<NetConncetionTypeEnum>(conncetion.ClientProvidedName, out var type))
                return type;
            else
                return NetConncetionTypeEnum.Unknown;
        }

        private async Task SubscribeAsync(IConnection cconn, IGrouping<string, SubscribeInfo> info, CancellationToken t)
        {
            if (!_isWorkQueue && !info.Any(m => !string.IsNullOrWhiteSpace(m.Topic)))
                return;
            var consumer = new ConsumerInfo();
            var channel = await CreateChannelAsync(cconn, t);
            await channel.QueueDeclareAsync(info.Key, true, false, false, null, cancellationToken: t);
            await channel.BasicQosAsync(0, _config.PrefetchCount, false, t);
            if (!_isWorkQueue)
            {
                await channel.ExchangeDeclareAsync(_exchangeName, ExchangeType.Topic, true, false, null, cancellationToken: t);
                foreach (var tempTopic in info)
                {
                    if (!string.IsNullOrWhiteSpace(tempTopic.Topic))
                    {
                        await channel.QueueBindAsync(info.Key, _exchangeName, tempTopic.Topic, null, cancellationToken: t);
                        if (!consumer.SubscribeInfos.Any(m => m.Group == info.Key && m.Topic == tempTopic.Topic))
                            consumer.SubscribeInfos.Add(new SubscribeInfo { Group = info.Key, Topic = tempTopic.Topic });
                    }
                }
            }
            else
            {
                if (!consumer.SubscribeInfos.Any(m => m.Group == info.Key))
                    consumer.SubscribeInfos.Add(new SubscribeInfo { Group = info.Key, Topic = info.Key });
            }

            consumer.Consumer = CreateConsumerAsyc(channel, t);
            await channel.BasicConsumeAsync(info.Key, false, consumer.Consumer, t);
            _consumers.Add(consumer);
        }

        private async Task UnsubscribeAsync(IGrouping<string, SubscribeInfo> info, CancellationToken t)
        {
            if (string.IsNullOrWhiteSpace(info.Key))
                return;
            var index = -1;
            foreach(var consumerInfo in _consumers.Where(m => m.SubscribeInfos.Exists(j => j.Group == info.Key)).ToList())
            {
                foreach(var unsubscribe in info)
                {
                    if (_isWorkQueue)
                        index = consumerInfo.SubscribeInfos.FindIndex(j => j.Group == info.Key);
                    else
                        index = consumerInfo.SubscribeInfos.FindIndex(j => j.Topic == unsubscribe.Topic);
                    if (index == -1)
                        continue;
                    consumerInfo.SubscribeInfos.RemoveAt(index);
                    if (!_isWorkQueue)
                        await consumerInfo.Consumer.Channel.QueueUnbindAsync(unsubscribe.Group, _exchangeName, unsubscribe.Topic, cancellationToken: t);
                    //else
                        // await consumerInfo.Consumer.Channel.QueueDeleteAsync(subscribe.Group, cancellationToken: t);
                }

                if (consumerInfo.SubscribeInfos.Count == 0)
                {
                    consumerInfo.Consumer.ReceivedAsync -= ConsumerReceiveEventAsync;
                    await consumerInfo.Consumer.Channel.BasicCancelAsync(consumerInfo.Consumer.ConsumerTags[0], cancellationToken: t);
                    //await consumerInfo.Consumer.Channel.CloseAsync(t);
                    await consumerInfo.Consumer.Channel.DisposeAsync();
                    consumerInfo.Consumer = null;
                }
            }

            foreach(var del in _consumers.Where(m => m.SubscribeInfos.Count == 0))
            {
                _consumers.Remove(del);
            }
        }

        private async Task PublishCoreAsync(string topic, byte[] data, CancellationToken token)
        {
            IConnection pConnection = null;
            if (_config.NewConnectionPerSend)
                pConnection = await CreateConnectionAsync(_config.Address, NetConncetionTypeEnum.Producer.ToString(), token);
            else
            {
                lock (this)
                {
                    pConnection = _pConnections[_curPIndex];
                    _curPIndex += 1;
                    if (_curPIndex >= _pConnections.Length)
                        _curPIndex = 0;
                }
            }
            try
            {
                using (var channel = await CreateChannelAsync(pConnection!, token))
                {
                    var properties = new BasicProperties();
                    var msg = new DispatcherMessage(topic, data, this);
                    if (_isWorkQueue)
                    {
                        await channel.QueueDeclareAsync(topic, true, false, false, null, cancellationToken: token);
                        if (_filterService != null)
                            await _filterService.PreSendAsync(msg);
                        await channel.BasicPublishAsync("", msg.Topic, false, properties, msg.Message, cancellationToken: token);
                        if (_filterService != null)
                            await _filterService.AfterSendAsync(msg);
                    }
                    else
                    {
                        await channel.ExchangeDeclareAsync(_exchangeName, ExchangeType.Topic, true, false, null, cancellationToken: token);
                        if (_filterService != null)
                            await _filterService.PreSendAsync(msg);
                        await channel.BasicPublishAsync(_exchangeName, msg.Topic, false, properties, msg.Message, cancellationToken: token);
                        if (_filterService != null)
                            await _filterService.AfterSendAsync(msg);
                    }
                }
            }
            finally
            {
                if (_config.NewConnectionPerSend)
                    pConnection?.Dispose();
            }
        }

        #endregion
    }
}
